/**
 * Copyright 2009 Takahiko Ito
 *
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.unigram.oluolu.rqe;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.unigram.oluolu.common.FsUtil;
import org.unigram.oluolu.common.OluoluConstants;
import org.unigram.oluolu.common.OluoluLogger;
import org.unigram.oluolu.common.Query;
import org.unigram.oluolu.rqe.postprocessor.IPostProcessor;

/**
 * Create related query dictionary from query log data. Users can 
 * create the classes to create the dictionary which they need such 
 * as spell check dictionary, synonym dictionary or context dictionary 
 * by implementing setSpecificConfiguration and the classes which are 
 * selected the setSpecificConfiguration method.
 */
public abstract class RelatedQueryDictionaryCreation 
    extends Configured implements Tool {

    /** logger. */
    private static OluoluLogger logger = OluoluLogger.getLogger();

    /**
     * Set task specific configuration such as selecting Selector
     * and PostProcessor. This method need to be implemented in
     * inherited classes.
     * 
     * Waning: user needs to extend this class implementing this method
     * see the implementation SpellCheckDictionaryCreation class as 
     * a example.
     *
     * @param conf Configuration object
     * @param args argument strings which containing specific 
     *         configuration information.
     * 
     */
    protected abstract void setSpecificConfiguration(
            Configuration conf, String[] args);

    /**
     * Show usage.
     * Warning: user needs to implement this method.
     */
    protected abstract void showParameters();
    
    /**
     * Show common parameters for RelatedQueryDictionaryCreation 
     * and the inherited classes.
     */
    protected void showCommonParamters() {
        System.out.println("Common paramters:");
        System.out.println("    -input INPUT                 " 
                + "use INPUT as input resource");
        System.out.println("    -output OUTPUT               " 
                + "use OUTPUT as outupt prefix");
        System.out.println("    [-timeRange TIME]            "
                + "use TIME as the threshold of the" 
                + "maximum time range in one session");
        System.out.println("    [-queriesPerPerson NUM]      " 
                + "use NUM as the threshold of "
                + "the maximum"
                + "number of queryies per person");
        System.out.println("    [-minimumSupport MIN]        "
                + "use MIN as the minimum frequency "
                + "of the query string");
        System.out.println("    [-postThreshold TH]        "
                + "use TH as the minimum frequency "
                + "of the queries stored in PostProcessors");        
        System.out.println("    [-showScore]                 "
                + "output the scores of elements of output   "
                + "dictionary"
                + "of the query string");        
        System.out.println("    [-help]                      "
                + "show usage");
    }
    
    /**
     * Extract related query pairs.
     * 
     * @param args argument strings containing inpu and 
     *          output information.
     * @param conf send to mapper and reducer the 
     *         configuration information.
     * @return 0 when succeeded
     * @throws IOException -
     * @throws InterruptedException -
     * @throws ClassNotFoundException -
     */
    public final int run(final String[] args,
            final Configuration conf)
    throws IOException, InterruptedException,
            ClassNotFoundException {
        
        this.setDefaultConfiguration();
        FileSystem fs = FileSystem.get(conf);

        String input = "";
        String outputPrefix = "";
        boolean showScore = false;

        for (int i = 0; i < args.length; ++i) {
            if ("-input".equals(args[i])) {
                input = args[++i];
            } else if ("-output".equals(args[i])) {
                outputPrefix = args[++i];
            } else if ("-timeRange".equals(args[i])) {
                conf.setLong(RqeConstants.RQE_TIME_RANGE, 
                        Long.parseLong(args[++i]));
            } else if ("-queriesPerPerson".equals(args[i])) {
                conf.setInt(RqeConstants.RQE_MAX_QUERY_NUMBER, 
                        Integer.parseInt(args[++i]));
            } else if ("-minimumSupport".equals(args[i])) {
                conf.setInt(RqeConstants.RQE_MINIMUM_SUPPORT, 
                        Integer.parseInt(args[++i]));
            } else if ("-postThreshold".equals(args[i])) {
            	conf.setInt(RqeConstants.POST_FREQUENCY_THRESHOLD, 
            			Integer.parseInt(args[++i]));
            } else if ("-showScore".equals(args[i])) {
                conf.setBoolean(RqeConstants.RQE_SHOW_SCORE, true);
            } else if ("-help".equals(args[i])) {
                this.showParameters();
                return 0;
            }
        }       
        
        String outputCandidate = outputPrefix + "."
            + conf.get(RqeConstants.RQE_CANDIDATE_SUFFIX,
                RqeConstants.DEFAULT_RQE_CANDIDATE_SUFFIX);
        String output = outputPrefix + "."
            +  conf.get(RqeConstants.RQE_SUFFIX,
                RqeConstants.DEFAULT_RQE_SUFFIX);
        
        /* output (validated in post-processor) */
        String tmpOutput = output + ".tmp";

        if (input.length() == 0) {
            logger.logSevere("Invalid input");
            return 1;
        } else if (outputPrefix.length() == 0) {
            logger.logSevere("Invalid output prefix");
            return 1;
        }
        
        this.setSpecificConfiguration(conf, args);
        this.reportSettings(conf);

        boolean result = extractCandidate(input, 
                outputCandidate, conf, fs);        

        this.narrowDownCandidates(outputCandidate, 
                tmpOutput, conf, fs); 
        
        this.runPostprocessor(
                conf.get(RqeConstants.RQE_POST_PROCESSOR),
                input, tmpOutput, output, conf, fs);

        /* clean temporal files */
        FsUtil.clean(fs, outputCandidate, tmpOutput);
        
        return 0;
    }


    /**
     * Flush current settings of Oluolu. This method can be override in
     * inherited classes to flash more setting.
     *
     * @param conf Configuration object
     */
    protected final void reportSettings(final Configuration conf) {

        String selectorName = conf.get(RqeConstants.RQE_SELECTOR);
       if (selectorName == null) {
           logger.logSevere("Selector is not selected.");
       } else {
           logger.logInfo("Selector: " +  selectorName + " is applied.");
       }

       String postProcessorName = conf.get(
               RqeConstants.RQE_POST_PROCESSOR);
       if (postProcessorName == null) {
           logger.logSevere("PostProcessor is not selected.");
       } else {
           logger.logInfo("PostProcessor: " + postProcessorName 
                   + " is applied.");
       }

       int numReducers = conf.getInt(
               OluoluConstants.NUMBER_OF_REDUCES,
                   OluoluConstants.DEFAULT_NUMBER_OF_REDUCES);
       logger.logInfo("Number of reducers: " + numReducers);

       return;
    }

    /**
     * Run post processor.
     * 
     * @param postProcessorClassName class name of applied post processor
     * @param input input file name
     * @param outputBeforeReduced temporal output file before post processor
     * @param output output file prefix
     * @param conf Configuration object
     * @param fs FileSystem
     */
    private void runPostprocessor(
            final String postProcessorClassName,
            final String input, final String outputBeforeReduced,
            final String output, final Configuration conf,
            final FileSystem fs) {

        IPostProcessor postProcessor = null;
        try {
            if (postProcessorClassName == null
                    || postProcessorClassName.length() == 0) {
                return;
            }
            Class<? extends IPostProcessor> postProcessorClass = Class.forName(
                     postProcessorClassName).asSubclass(IPostProcessor.class);
             Constructor<? extends IPostProcessor> constructor =
                 postProcessorClass.getConstructor(Configuration.class);
             postProcessor = constructor.newInstance(conf);
         } catch (NoSuchMethodException nsme) {
             throw new RuntimeException(nsme);
         } catch (ClassNotFoundException cnfe) {
             throw new RuntimeException(cnfe);
         } catch (InstantiationException ie) {
             throw new RuntimeException(ie);
         } catch (IllegalAccessException iae) {
             throw new RuntimeException(iae);
         } catch (InvocationTargetException ite) {
             throw new RuntimeException(ite.getCause());
         }
         logger.logInfo("starting PostProcessor...");
         postProcessor.run(input, outputBeforeReduced,
                 output);
    }

    /**
     * Set default configurations.
     */
    private void setDefaultConfiguration()  {
        Configuration.addDefaultResource("conf/oluolu-default.xml");
        Configuration.addDefaultResource("conf/oluolu-site.xml");
        return;
    }

    /**
     * Main Procedure.
     * 
     * @param args argument string containing set up information 
     *          such as input and output 
     * @return int return 0 succeeded
     * @throws IOException -
     * @throws InterruptedException -
     * @throws ClassNotFoundException -
     */
    public final int run(final String[] args) 
    throws IOException,
        InterruptedException, ClassNotFoundException {
        Configuration conf = getConf();
        return run(args, conf);
    }

    /**
     * Extract candidate query pairs which may be related to each other.
     *
     * @param input input file name
     * @param output output file prefix
     * @param conf Configuration obcjet
     * @param fs FileSystem
     * @return true when succeeded otherwise return false
     * @throws IOException -
     * @throws InterruptedException -
     * @throws ClassNotFoundException -
     */
    private boolean extractCandidate(final String input,
            final String output, final Configuration conf,
            final FileSystem fs)
        throws IOException, InterruptedException,
            ClassNotFoundException {

        Path inputPath = new Path(input);
        Path outputPath = new Path(output);
        FsUtil.checkPath(outputPath, fs);

        Job job = new Job(conf);
        job.setJarByClass(RelatedQueryDictionaryCreation.class);
        FileInputFormat.addInputPath(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);
        job.setMapperClass(RelatedCandidateExtractionMapper.class);
        job.setReducerClass(RelatedCandidateExtractionReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Query.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        job.setNumReduceTasks(conf.getInt(OluoluConstants.NUMBER_OF_REDUCES,
                OluoluConstants.DEFAULT_NUMBER_OF_REDUCES));
        // SequenceFileOutputFormat.setOutputCompressorClass(job,
        // GzipCodec.class);
        // SequenceFileOutputFormat.setOutputCompressionType(job,
        // CompressionType.BLOCK);
        boolean result = job.waitForCompletion(true);
        this.setResultConf(job.getCounters(), conf);
        return result;
    }
    
    /**
     * Append result log from extractCandidates into conf.
     *
     * @param counters contains log information
     * @param conf Configuration
     */
    private void setResultConf(final Counters counters, 
            final Configuration conf) {
        conf.setLong(OluoluConstants.OLUOLU_INPUT_RECORDS, 
                counters.findCounter(
                        OluoluConstants.COUNTER_GROUP, 
                   "MAP_INPUT_RECORDS").getValue());
    }

    /**
     * Extract candidate query pairs only the ones which are
     * validated selector.
     *
     * @param input input file name
     * @param output output file prefix
     * @param conf Configuration
     * @param fs FileSytem
     * @return boolean
     * @throws IOException -
     * @throws InterruptedException -
     * @throws ClassNotFoundException -
     */
    private Counters narrowDownCandidates(final String input,
            final String output, final Configuration conf,
            final FileSystem fs)
        throws IOException, InterruptedException,
            ClassNotFoundException {
        Path inputPath = new Path(input);
        Path outputPath = new Path(output);
        FsUtil.checkPath(outputPath, fs);

        Job job = new Job(conf);
        FileInputFormat.addInputPath(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);
        job.setJarByClass(RelatedQueryDictionaryCreation.class);
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setMapperClass(RelatedQueryExtractionMapper.class);
        job.setReducerClass(RelatedQueryExtractionReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        job.setNumReduceTasks(conf.getInt(OluoluConstants.NUMBER_OF_REDUCES,
                OluoluConstants.DEFAULT_NUMBER_OF_REDUCES));
        // SequenceFileOutputFormat.setOutputCompressorClass(job,
        // GzipCodec.class);
        // SequenceFileOutputFormat.setOutputCompressionType(job,
        // CompressionType.BLOCK);
        job.waitForCompletion(true);
        return job.getCounters();
    }
}
